# -*- coding: UTF-8 -*-
"""
@auth:bxj
@date:2019-08-09
@describe:获取数据库中信息
"""
import json
import time

from typing import Optional, Any, List

import pandas as pd
import pymysql

from app.common.common_func import Common, Encoder
from app.config.configer import Config
from app.log.ulog import logger_
from pymysql.err import InternalError
from sqlalchemy import create_engine, text


# noinspection SqlResolve
class GetSqlData:

	@staticmethod
	def conn_database(environment, source: Optional[str] = "saas"):
		"""
		数据库连接
		:param source: str
		:type environment:str
		"""
		yaml_data = Common.get_yaml_data("config", "database.yaml")
		config = yaml_data[environment][source]
		config["cursorclass"] = pymysql.cursors.DictCursor
		try:
			connect = pymysql.connect(**config)
			return connect
		except Exception:
			raise

	@staticmethod
	def exec_update(
			environment: str,
			sql: str,
			source: Optional[str] = "saas"
	):
		"""执行更新语句"""
		conn = None
		cur = None
		try:
			conn = GetSqlData.conn_database(environment, source)
			cur = conn.cursor()
			cur.execute(sql)
			conn.commit()
			logger_.info(f"执行sql:{sql}")
		except Exception:
			conn.rollback()
			raise
		finally:
			cur.close()
			conn.close()

	@staticmethod
	def exec_select(
			environment: str,
			sql: str,
			source: Optional[str] = "saas"
	) -> List:
		"""执行查询语句"""
		conn = None
		cur = None
		try:
			conn = GetSqlData.conn_database(environment, source)
			cur = conn.cursor()
			logger_.info(f"""执行查询语句:{sql}""")
			cur.execute(sql)
			fetch = cur.fetchall()
			logger_.info(f"""查询结果为:{fetch}""")
			if len(fetch) > 0:
				return fetch
			else:
				return []
		except InternalError:
			raise
		except Exception:
			cur.close()
			conn.close()
			logger_.exception("系统异常!")

	@staticmethod
	def get_sub_table(environment: str, asset_id: str) -> str:
		table = None
		if environment == "test":
			table = str((int(asset_id) % 8 + 1))
		elif environment in ("qa", "prod"):
			table = str((int(asset_id) >> 22) % 8 + 1)
		return table

	@staticmethod
	def check_credit_step(environment: str, credit_id: str) -> str:
		"""
		检查授信步骤
		"""
		sql = f'''Select credit_step from sandbox_saas.credit where id = {credit_id};'''
		credit_step = GetSqlData.exec_select(environment, sql)[0].get("credit_step")
		return credit_step

	@staticmethod
	def change_credit_status(environment: str, credit_id: str) -> str:
		"""修改授信表状态与步骤"""
		create_time = Common.get_new_time("before", "minutes", 10)
		sql = f"""update sandbox_saas.credit set create_time='{create_time}' where id='{credit_id}';"""
		GetSqlData.exec_update(environment, sql)

	@staticmethod
	def credit_set(environment: str, credit_id: str, credit_amount: str = 500000, stat: str = "approve") -> str:
		"""授信时调用，根据环境判断是否需要等待补偿"""
		logger_.info("开始检查授信步骤")
		if Config().get_item("Switch", "credit") == "1":
			logger_.info("根据配置识别当前授信走虚拟风控流程")
			# GetSqlData.change_credit_step(environment, credit_id)
			GetSqlData.change_credit_status(environment, credit_id)
			GetSqlData.change_athena_status(environment, credit_id, credit_amount=credit_amount, stat=stat)
			Common.trigger_task(job_name="creditReparationJob", env=environment)
		else:
			logger_.info("根据配置识别当前授信走真实风控流程")
		status = 2
		version = 1
		while status != 4:
			if version > 20:
				logger_.info("授信未成功")
				break
			step = GetSqlData().check_credit_step(environment, credit_id)
			if step != 4:
				logger_.info(f"当前授信步骤为:{step};当前循环次数为:{version}")
				Common.trigger_task(job_name="creditReparationJob", env=environment)
				version += 1
				time.sleep(2)
			elif step == 4:
				logger_.info("当前授信已完成,可以进行下个步骤!")
				status = 4

	@staticmethod
	def check_loan_result(environment: str, project_id: str) -> str:
		"""查询放款状态"""
		sql = f"""Select loan_result from sandbox_saas.project_detail where id = '{project_id}';"""
		loan_result = GetSqlData.exec_select(environment, sql)[0].get('loan_result')
		return loan_result

	@staticmethod
	def loan_set(environment: str, project_id: str) -> str:
		"""放款申请后调用，查询放款状态是否成功"""
		logger_.info("开始检查放款步骤")
		# datas = '{"projectId":"%s","code":2000,"success":true,"inProcess":false}'%project_id
		# Ulog().logger_().info(datas)
		# rep = requests.post(url="http://api-qa1.cloudloan.com:9011/api/v1/busi/callback/loan/apply",
		# 			  data=datas)
		# Ulog().logger_().info(rep.status_code)
		# Ulog().logger_().info(rep.text)
		# http://api-qa1.cloudloan.com 39.107.43.201
		# while True:
		Common.trigger_task("projectLoanReparationJob", environment)
		if GetSqlData().check_loan_result(environment, project_id) == -1:
			raise Exception("放款状态不正确，未申请放款成功")
		else:
			try:
				version = 1
				while GetSqlData().check_loan_result(environment, project_id) != 1:
					if version > 15:
						logger_.info(f"循环{version - 1}次未查询到放款成功状态，判断为放款失败")
						break
					res = GetSqlData().check_loan_result(environment, project_id)
					if res == 0:
						logger_.info(f"当前loan_result为:{res};放款失败")
						break
					if res != 1:
						logger_.info(f"当前loan_result为:{res};当前循环次数为:{version}")
						Common.trigger_task("projectLoanReparationJob", environment)
						version += 1
						time.sleep(2)
			except Exception as e:
				raise e

	@staticmethod
	def check_pay_order_code(environment: str, project_id: str):
		"""检查steamrunner.pay_order的code"""
		sql = f"""Select code from sandbox_saas_steamrunner.sr_pay_order where project_id = {project_id};"""
		try:
			code = GetSqlData.exec_select(environment, sql)[0].get("code", None)
			return code
		except TypeError:
			return None

	@staticmethod
	def change_pay_status(environment: str, project_id: str):
		"""修改steamrunner.pay_order的放款状态为成功"""
		finish_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
		# finish_time = "2021-03-29 00:00:00"
		if Config().get_item("Switch", "loan") == '1':
			logger_.info("放款开关已关闭，走虚拟放款逻辑")
			sql1 = f"""
				Update sandbox_saas_steamrunner.sr_pay_order 
				set code=2000,msg='成功',finish_time='{finish_time}' 
				where project_id={project_id};
				"""
			sql2 = f"""
				Update sandbox_saas.project_loan_flow
				set loan_result=2
				where project_id={project_id};
				"""
			sql3 = f"""
				Update sandbox_saas.project_loan_record
				set loan_result=2 
				where project_id={project_id};
				"""
			sql4 = f"""
				Update sandbox_saas.project_detail
				set loan_result=2,loan_status=0,loan_step=0
				where id={project_id};
				"""
			loop = 0
			while GetSqlData.check_pay_order_code(environment, project_id) is None:
				if loop > 100:
					raise KeyError("steam runner未生成放款数据")
				loop += 1
			if GetSqlData.check_pay_order_code(environment, project_id) in (2002, 2003):
				sqls = [sql1, sql2, sql3, sql4]
			else:
				sqls = [sql1]
			for sql in sqls:
				GetSqlData.exec_update(environment, sql)
		else:
			logger_.info("放款开关已开启,走真实放款流程")

	@staticmethod
	def get_asset_id(environment: str, project_id: str) -> str:
		"""获取资产id"""
		sql = f"""select id from asset WHERE project_id='{project_id}';"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			asset_id = rep[0].get("id")
			return asset_id
		else:
			logger_.info("未查询到资产ID")

	@staticmethod
	def get_user_repayment_detail(
			project_id: str,
			environment: str,
			period: int,
			repayment_plan_type: str,
			fee_category: int = 2002
	) -> dict:
		"""
		获取用户还款计划中的关联
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		if repayment_plan_type in ["1", "2"]:
			plan_table = 'user_repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
			sql = f"""
				select * from {plan_table}
				where asset_id = {asset_id}
				and period = {period}
				and repayment_plan_type = {repayment_plan_type};
				"""
		elif fee_category == 3003:
			plan_table = 'user_fee_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
			sql = f"""
				select * from {plan_table} 
				where asset_id = {asset_id} 
				and period = {period} 
				and fee_category = {repayment_plan_type};
				"""
		else:
			plan_table = 'fee_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
			sql = f"""
				select plan_pay_date,rest_amount,cur_amount,source_plan_id from {plan_table} 
				where asset_id = {asset_id} 
				and period = {period} 
				and fee_category = {fee_category};
				"""
		return json.loads(json.dumps(GetSqlData.exec_select(environment, sql)[0], cls=Encoder))

	@staticmethod
	def get_user_repayment_amount(project_id: str, environment: str, period: str) -> float:
		"""获取用户还款计划当期应还款总额"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'user_repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""
			select sum(cur_amount) as cur_amount from {plan_table} 
			where asset_id={asset_id} 
			and period={period};
			"""
		return float(GetSqlData.exec_select(environment, sql)[0].get('cur_amount'))

	@staticmethod
	def get_repayment_amount(project_id: str, environment: str, period: str) -> float:
		"""获取机构还款计划当期应还款总额"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""
			select sum(cur_amount) as amount from {plan_table} 
			where asset_id='{asset_id}' and period={period};
			"""
		return float(GetSqlData.exec_select(environment, sql)[0].get("amount"))

	@staticmethod
	def get_credit(environment: str, credit_id: Optional[str] = None, source_user_id: Optional[str] = None) -> dict:
		"""获取授信表"""
		if (credit_id is None) and (source_user_id is None):
			raise ValueError("授信ID或来源用户ID不能为空")
		elif (credit_id is not None) and (source_user_id is not None):
			raise ValueError("授信ID与来源用户ID只能传一个")
		if credit_id is not None:
			sql = f"""
					select * from credit where id={credit_id};
					"""
		elif source_user_id is not None:
			sql = f"""
						select * from credit where source_user_id='{source_user_id}';
						"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			credit = json.loads(json.dumps(rep[0], cls=Encoder))
			return credit

	@staticmethod
	def get_credit_customer(environment: str, credit_id: str) -> dict:
		"""获取授信用户表"""
		sql = f"""
				select * from credit_customer where credit_id={credit_id};
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			credit_customer = json.loads(json.dumps(rep[0], cls=Encoder))
			return credit_customer

	@staticmethod
	def get_credit_entity(environment: str, credit_id: str) -> dict:
		"""获取授信实体表"""
		sql = f"""
				select * from credit_entity where credit_id={credit_id};
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			credit_entity = json.loads(json.dumps(rep[0], cls=Encoder))
			return credit_entity

	@staticmethod
	def get_credit_extra(environment: str, credit_id: str) -> dict:
		"""获取授信扩展表"""
		sql = f"""
				select * from credit_extra where credit_id={credit_id};
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			credit_extra = json.loads(json.dumps(rep[0], cls=Encoder))
			return credit_extra

	@staticmethod
	def get_all_repayment_amount(project_id: str, environment: str) -> float:
		"""获取资产还款金额"""
		sql = f"""select amount from asset where project_id='{project_id}';"""
		return float(GetSqlData.exec_select(environment, sql)[0].get('amount'))

	@staticmethod
	def get_maturity(project_id: str, environment: str) -> str:
		"""获取资产期数"""
		sql = f'''select maturity from asset where project_id="{project_id}";'''
		return GetSqlData.exec_select(environment, sql)[0].get("maturity")

	@staticmethod
	def get_debt_amount(project_id, environment):
		"""获取资产剩余应还本金"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		sql = f"""select debt_amount from asset where id={asset_id};"""
		return GetSqlData.exec_select(environment, sql)[0].get('debt_amount')

	@staticmethod
	def get_repayment_plan_date(
			project_id: str,
			environment: str,
			repayment_plan_type: str,
			period: str
	) -> dict:
		"""
		获取某期机构还款计划
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""
			select * 
			from {plan_table} 
			where asset_id = {asset_id}
			and period = {period} 
			and repayment_plan_type = {repayment_plan_type};
			"""
		return json.loads(json.dumps(GetSqlData.exec_select(environment, sql)[0], cls=Encoder))

	@staticmethod
	def get_repayment_plan(
			project_id: str,
			environment: str,
			period: Optional[str] = None
	) -> list:
		"""
		获取机构还款计划
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		if period is None:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id};
					"""
		else:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id}
					and period={period};
					"""
		plan = json.loads(json.dumps(GetSqlData.exec_select(environment, sql), cls=Encoder))
		return plan

	@staticmethod
	def get_fee_plan(
			project_id: str,
			environment: str,
			period: Optional[str] = None
	) -> list:
		"""
		获取费计划
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'fee_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		if period is None:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id};
					"""
		else:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id}
					and period={period};
					"""
		plan = json.loads(json.dumps(GetSqlData.exec_select(environment, sql), cls=Encoder))
		return plan

	@staticmethod
	def get_project_detail(environment: str, project_id: str) -> dict:
		"""获取进件表"""
		sql = f"""
				select * from project_detail where id='{project_id}';
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			project_detail = json.loads(json.dumps(rep[0], cls=Encoder))
			return project_detail

	@staticmethod
	def get_project_customer_detail(environment: str, project_id: str) -> dict:
		"""获取进件用户表"""
		sql = f"""
				select * from project_customer_detail where project_id={project_id};
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			project_customer_detail = json.loads(json.dumps(rep[0], cls=Encoder))
			return project_customer_detail

	@staticmethod
	def get_project_entity_detail(environment: str, project_id: str) -> dict:
		"""获取进件实体表"""
		sql = f"""
				select * from project_entity_detail where project_id='{project_id}';
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			project_entity_detail = json.loads(json.dumps(rep[0], cls=Encoder))
			return project_entity_detail

	@staticmethod
	def get_project_extra_detail(environment: str, project_id: str) -> dict:
		"""获取进件扩展表"""
		sql = f"""
						select * from project_extra_detail where project_id={project_id};
						"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			project_extra_detail = json.loads(json.dumps(rep[0], cls=Encoder))
			return project_extra_detail

	@staticmethod
	def get_repayment(
			project_id: str,
			environment: str,
	) -> list:
		"""
		获取还款表
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		sql = f"""
				select * 
				from repayment
				where asset_id = {asset_id}
				order by create_time;
				"""
		repayment = json.loads(json.dumps(GetSqlData.exec_select(environment, sql), cls=Encoder))
		return repayment

	@staticmethod
	def get_repayment_detail(
			project_id: str,
			environment: str,
			period: Optional[str] = None
	) -> list:
		"""
		获取还款详情表
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'repayment_detail_0' + GetSqlData.get_sub_table(environment, asset_id)
		if period is None:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id};
					"""
		else:
			sql = f"""
					select * 
					from {plan_table} 
					where asset_id = {asset_id}
					and period={period};
					"""
		plan = json.loads(json.dumps(GetSqlData.exec_select(environment, sql), cls=Encoder))
		return plan

	@staticmethod
	def get_swap_detail(
			project_id: str,
			environment: str,
	) -> list:
		"""
		获取债转详情表
		"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		sql = f"""
				select * 
				from asset_swap_detail
				where asset_id = {asset_id};
				"""
		swap_detail = json.loads(json.dumps(GetSqlData.exec_select(environment, sql), cls=Encoder))
		return swap_detail

	@staticmethod
	def check_user_amount(user_id, environment):
		"""查询用户可用额度"""
		version = 1
		while True:
			if version > 10:
				logger_.info(f'''当前查询次数{version}''')
				break
			user_amount = GetSqlData.user_amount(user_id, environment)
			if user_amount > 0.00:
				logger_.info("额度检查完成")
				break
			elif user_amount == 0.000000:
				version += 1
				time.sleep(5)

	@staticmethod
	def user_amount(user_id: str, environment: str):
		"""查询用户可用额度"""
		sql = f'select available_amount from amount where user_id={user_id};'
		return GetSqlData.exec_select(environment, sql, "nebula")[0].get('available_amount')

	@staticmethod
	def select_user_amount(environment: str, source_user_id: str) -> dict:
		"""查询用户额度表数据"""
		sql = f"""select * from amount where source_user_id='{source_user_id}';"""
		rep = GetSqlData.exec_select(environment, sql, "nebula")
		if rep:
			amount = json.loads(json.dumps(rep[0], cls=Encoder))
			return amount
		else:
			logger_.info("未查询到用户额度记录")

	@staticmethod
	def project_audit_status(project_id: str, environment: str) -> int:
		"""查询进件审核状态"""
		sql = f"""select audit_status from project_detail where id={project_id};"""
		audit_status = GetSqlData.exec_select(environment, sql)[0].get('audit_status')
		return audit_status

	@staticmethod
	def project_result(project_id: str, environment: str) -> None:
		"""进件审核结果查询"""
		logger_.info("开始检查进件审核步骤")
		try:
			version = 1
			while True:
				if version > 10:
					logger_.info(f"{version}次未查询到进件审核成功状态")
					break
				audit_status = GetSqlData().project_audit_status(project_id, environment)
				if audit_status != 2:
					logger_.info(f"当前进件审核状态为:{audit_status};当前查询次数为:{version}")
					version += 1
					time.sleep(10)
				elif audit_status == 2:
					logger_.info("进件审核成功")
					break
		except Exception:
			raise

	@staticmethod
	def change_project_audit_status(project_id: str, environment: str, stat: str = "approve"):
		"""修改进件审核状态为通过"""
		if Config().get_item("Switch", "project") == '1':
			logger_.info("风控已关闭，走虚拟进件风控逻辑")
			if stat == "approve":
				sql = f"""
						UPDATE risk_apply
						SET audit_result = 'APPROVE', audit_message = '风控审核通过',
						return_code = 2000, step = 'COMPLETED'
						WHERE apply_id = '{project_id}';
					"""
			elif stat == "reject":
				sql = f"""
						UPDATE risk_apply
						SET audit_result = 'REJECT', audit_message = '风控审核拒绝',
						return_code = 2000, step = 'COMPLETED'
						WHERE apply_id = '{project_id}';
					"""
			else:
				raise ValueError(f"无效状态:{stat}")
			GetSqlData.exec_update(environment, sql, "athena")
			Common.trigger_task("projectReparationJob", environment)
		else:
			logger_.info("风控开关已开启，走真实风控流程")
			GetSqlData.project_result(project_id, environment)

	@staticmethod
	def loan_sql(env: str):
		"""修改steamrunner放款状态"""
		sql1 = f"""UPDATE sandbox_saas.project_loan_record
					SET loan_result = 2
					WHERE project_id IN (
							SELECT id
							FROM sandbox_saas.project_detail
							WHERE project_detail.loan_result != 1
					)
					AND loan_result = 0;"""
		sql2 = f"""UPDATE sandbox_saas_steamrunner.sr_pay_order
					SET code = 2000
					WHERE project_id IN (
							SELECT id
							FROM sandbox_saas.project_detail
							WHERE project_detail.loan_result != 1
						);"""
		sql3 = f"""UPDATE sandbox_saas.project_detail
				SET loan_result = 2
				WHERE id IN (
						SELECT a.project_id
						FROM (
							SELECT project_id
							FROM sandbox_saas.project_loan_flow
							WHERE project_id IN (
								SELECT id
								FROM sandbox_saas.project_detail
								WHERE (loan_result != 1
									AND audit_result = 1)
							)
						) a
					);"""
		sql4 = f"""UPDATE sandbox_saas.project_loan_flow
					SET loan_result = 2
					WHERE loan_result != 1;"""
		sql_list = [sql1, sql2, sql3, sql4]
		for sql in sql_list:
			GetSqlData.exec_update(environment=env, sql=sql)

	@staticmethod
	def select_asset() -> int:
		"""查询没有放款成功的进件数量"""
		sql = f"""SELECT COUNT(DISTINCT project_id) AS c
					FROM sandbox_saas.project_loan_flow
					WHERE project_id IN (
						SELECT id
						FROM sandbox_saas.project_detail
						WHERE loan_result != 1
					);
				"""
		return GetSqlData.exec_select("qa", sql)[0].get("c")

	@staticmethod
	def get_current_period(project_id: str, environment: str) -> int:
		"""获取资产当前发生期"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		plan_table = 'repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""select period 
				from sandbox_saas.{plan_table} 
				where asset_id={asset_id} 
				and repayment_status=1 
				order by period limit 1
				"""
		period = GetSqlData.exec_select(environment, sql)[0].get('period')
		return period

	@staticmethod
	def get_asset(project_id: str, environment: str) -> dict:
		"""获取资产表内容"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		sql = f"""
				select * from asset
				where id={asset_id};
				"""
		asset = json.loads(json.dumps(GetSqlData.exec_select(environment, sql)[0], cls=Encoder))
		return asset

	@staticmethod
	def asset_count(environment: str):
		"""查询资产ID"""
		sql = "select id from asset;"
		asset_id = GetSqlData.exec_select(environment, sql)
		ids = []
		for i in asset_id:
			ids.append(i.get('id'))
		return ids

	@staticmethod
	def repayment_plan(asset_id: str, environment: str) -> int:
		"""查询机构还款计划条数"""
		plan_table = 'repayment_plan_0' + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""select count(*) as c from {plan_table} where asset_id={asset_id};"""
		count = GetSqlData.exec_select(environment, sql)[0].get('c')
		return count

	@staticmethod
	def select_need_audit_project_ids() -> list:
		"""
		查询要修改审核状态的进件ID
		用于Job
		"""
		sql = f"""SELECT id
				FROM project_detail
				WHERE product_code IN (
						'XJ_YLKJ_GFDDD'
					)
					AND audit_status != 2"""
		ids = GetSqlData.exec_select("qa", sql)
		idss = []
		for i in ids:
			idss.append(i.get("id"))
		return idss

	@staticmethod
	def select_need_sign_project_ids() -> list:
		sql = f"""
				Select id from project_detail
				where sign_status != 2 and loan_step != 4;
				"""
		ids = GetSqlData.exec_select("qa", sql)
		idss = []
		for i in ids:
			idss.append(i.get("id"))
		return idss

	@staticmethod
	def change_sign():
		sql1 = f"""
					update project_detail 
					set sign_status=2,sign_result=1,sign_time='{Common.get_time("-")}'
					where loan_step != 4;
					"""
		GetSqlData.exec_update(environment="qa", sql=sql1)

	@staticmethod
	def change_athena_status(environment: str, apply_id: str, credit_amount: int = 500000, stat: str = "approve"):
		"""修改Athena数据状态"""
		if stat == "approve":
			sql = f"""
					update sandbox_saas_athena.risk_apply
					set audit_result='APPROVE',quota={credit_amount},step='COMPLETED',return_code=2000,
					audit_message = '风控审核通过'
					where apply_id='{apply_id}';
					"""
		elif stat == "reject":
			sql = f"""
					update sandbox_saas_athena.risk_apply
					set audit_result='REJECT',quota={credit_amount},step='COMPLETED',return_code=2000,
					audit_message = '风控审核拒绝'
					where apply_id='{apply_id}';
					"""
		else:
			raise ValueError(f"无效状态:{stat}")
		GetSqlData.exec_update(environment, sql)

	@staticmethod
	def change_plan_pay_date(environment: str, project_id: str, period: int, date: str):
		"""修改还款计划应还时间"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		table = "repayment_plan_0" + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""
					update sandbox_saas.{table}
					set plan_pay_date='{date}'
					where asset_id={asset_id}
						and period={period}
						and repayment_status=1;
				"""
		GetSqlData.exec_update(environment, sql)

	@staticmethod
	def change_fee_plan_pay_date(environment: str, project_id: str, period: int, date: str):
		"""修改费计划应还时间"""
		asset_id = GetSqlData.get_asset_id(environment, project_id)
		table = "fee_plan_0" + GetSqlData.get_sub_table(environment, asset_id)
		sql = f"""
					update sandbox_saas.{table}
					set plan_pay_date='{date}'
					where asset_id={asset_id}
						and period={period}
						and fee_status=1
						and fee_category=9001;
				"""
		GetSqlData.exec_update(environment, sql)

	@staticmethod
	def change_asset(environment: str, project_id: str, values: [list]) -> None:
		"""修改资产表字段"""
		value = ""
		for v in values:
			value = value + v[0] + "=" + "'" + str(v[1]) + "'" + ","
		sql = f"""
					update sandbox_saas.asset 
					set {value.strip(",")}
					where sandbox_saas.asset.project_id='{project_id}'
				"""
		GetSqlData.exec_update(environment, sql)

	@staticmethod
	def select_entity_info(environment: str, busi_type: str, busi_id: str) -> dict:
		"""查询extra_info数据"""
		if busi_type == "credit":
			sql = f"""
					select extra_info from sandbox_saas.credit_entity where credit_id={busi_id};
				   """
		elif busi_type == "project":
			sql = f"""
					select extra_info from sandbox_saas.project_entity_detail where project_id={busi_id};
				   """
		else:
			raise KeyError("不支持的业务类型")
		extra_info = json.loads(GetSqlData.exec_select(environment, sql)[0].get("extra_info"))
		return extra_info

	@staticmethod
	def select_customer_info(environment: str, busi_type: str, busi_id: str) -> dict:
		"""查询customer_info数据"""
		if busi_type == "credit":
			sql = f"""
					select extra_info from sandbox_saas.credit_customer where credit_id={busi_id};
				   """
		elif busi_type == "project":
			sql = f"""
					select extra_info from sandbox_saas.project_customer_detail where project_id={busi_id};
				   """
		else:
			raise KeyError("不支持的业务类型")
		extra_info = json.loads(GetSqlData.exec_select(environment, sql)[0].get("extra_info"))
		return extra_info

	@staticmethod
	def get_file_download_record(environment: str, ass_id: str):
		"""查询文件下载记录表数据"""
		sql = f"""
				select * from sandbox_saas.file_download_record where ass_id={ass_id};
				"""
		rep = GetSqlData.exec_select(environment, sql)
		if rep is not None:
			record = json.loads(json.dumps(rep, cls=Encoder))
			return record
		else:
			return None

	@staticmethod
	def select_overdue_job_excute_status(environment: str):
		"""查询逾期任务执行状态"""
		if environment == "qa":
			sql = """
					select is_success from sandbox_saas.JOB_EXECUTION_LOG 
					where job_name='overdueForCloudloanJob'
					order by start_time
					desc limit 2;
			"""
		else:
			sql = """
					select is_success from sandbox_saas.job_execution_log
					where job_name='overdueForCloudloanJob'
					order by start_time
					desc limit 2;
			"""
		is_success = json.loads(GetSqlData.exec_select(environment, sql), cls=Encoder)
		return is_success

	@staticmethod
	def move_prod_data_to_test(environment: str, asset_id_list: List[str]):
		# 创建生产环境链接
		if environment != "qa":
			raise ValueError(f"暂不支持该环境:{environment}")
		yaml_data = Common.get_yaml_data("config", "database.yaml")
		prod_config = yaml_data["prod"]["saas"]
		logger_.info("链接生产环境mysql")
		prod_engine = create_engine(
			f"mysql+pymysql://{prod_config['user']}:{prod_config['password']}@{prod_config['host']}:{prod_config['port']}/{prod_config['db']}")
		test_config = yaml_data[environment]["saas"]
		logger_.info(f"链接{environment}环境mysql")
		test_engine = create_engine(
			f"mysql+pymysql://{test_config['user']}:{test_config['password']}@{test_config['host']}:{test_config['port']}/{test_config['db']}")
		table = (
			"asset_extra", "asset_fee", "asset_swap_apply", "asset_swap_detail",
			"overdue", "relief_apply", "relief_detail", "repayment"
		)
		# 同步资产表
		sql = f"""
			select * from asset where 
			id in {f"('" + "','".join(asset_id_list) + "')"};
		"""
		logger_.info("同步asset表数据————————————————————————————————")
		pdata = pd.read_sql(text(sql), prod_engine.connect())
		pdata.to_sql("asset", test_engine, if_exists="append", index=False)
		# 同步其他根据asset_id查询的表
		for t in table:
			sql = f"""
				select * from {t} where
				asset_id in {f"('" + "','".join(asset_id_list) + "')"};
			"""
			logger_.info(f"同步{t}表数据————————————————————————————————")
			pdata = pd.read_sql(text(sql), prod_engine.connect())
			pdata.to_sql(t, test_engine, if_exists="append", index=False)
		# 分表插入
		sub_table = (
			"fee_plan", "repayment_detail", "repayment_plan", "overdue_detail",
			"user_repayment_plan", "user_fee_plan"
		)
		for aid in asset_id_list:
			sub = GetSqlData.get_sub_table(environment, aid)
			for ta in sub_table:
				sql = f"""
					select * from {ta}_0{sub} where
					asset_id = '{aid}';
				"""
				logger_.info(f"同步{ta}_0{sub}表数据————————————————————————————————")
				pdata = pd.read_sql(text(sql), prod_engine.connect())
				pdata.to_sql(f'{ta}_0{sub}', test_engine, if_exists="append", index=False)
		# 进件相关信息插入
		project_table = (
			"project_customer_detail", "project_enterprise_detail",
			"project_entity_detail", "project_extra_detail", "repayment_binding_card"
		)
		# 进件详情表插入
		sql = f"""
				select * from project_detail where
				id in 
				(select project_id from asset where id in {f"('" + "','".join(asset_id_list) + "')"});
			"""
		logger_.info("同步project_detail表数据————————————————————————————————")
		pdata = pd.read_sql(text(sql), prod_engine.connect())
		pdata.to_sql('project_detail', test_engine, if_exists="append", index=False)
		# 进件其他表插入
		for tb in project_table:
			sql = f"""
				select * from {tb} where
				project_id in
				(select project_id from asset where id in {f"('" + "','".join(asset_id_list) + "')"});
			"""
			logger_.info(f"同步{tb}表数据————————————————————————————————")
			pdata = pd.read_sql(text(sql), prod_engine.connect())
			pdata.to_sql(tb, test_engine, if_exists="append", index=False)

		# 授信相关表插入
		credit_table = (
			"credit_customer", "credit_entity", "credit_extra"
		)
		# 授信申请表插入
		sql = f"""
			select * from credit where
			user_id in (select source_user_id from project_detail where id in (
			select project_id from asset where id in {f"('" + "','".join(asset_id_list) + "')"})
			);
		"""
		logger_.info("同步credit表数据————————————————————————————————")
		pdata = pd.read_sql(text(sql), prod_engine.connect())
		pdata.to_sql("credit", test_engine, if_exists="append", index=False)
		for tc in credit_table:
			sql = f"""
				select * from {tc} where credit_id in (
				(select source_user_id from project_detail where id in (
			select project_id from asset where id in {f"('" + "','".join(asset_id_list) + "')"})
			));
			"""
			logger_.info(f"同步{tc}表数据————————————————————————————————")
			pdata = pd.read_sql(text(sql), prod_engine.connect())
			pdata.to_sql(tc, test_engine, if_exists="append", index=False)

		# nebula相关信息同步
		sql = f"""
			select user_id from project_detail where id in (
			select project_id from asset where id in {f"('" + "','".join(asset_id_list) + "')"});
		"""
		res = GetSqlData.exec_select(environment, sql)
		if res:
			res = [str(x["user_id"]) for x in res]
			nebula_table = ["amount_detail", "amount"]
			nebula_config = yaml_data["prod"]["nebula"]
			nebula_engine = create_engine(
				f"mysql+pymysql://{nebula_config['user']}:{nebula_config['password']}@{nebula_config['host']}:{nebula_config['port']}/{nebula_config['db']}"
			)
			test_nebula_config = yaml_data[environment]["nebula"]
			test_nebula_engine = create_engine(
				f"mysql+pymysql://{test_nebula_config['user']}:{test_nebula_config['password']}@{test_nebula_config['host']}:{test_nebula_config['port']}/{test_nebula_config['db']}"
			)
			# 用户信息表同步
			sql = f"""
				select * from user_info where id in {f"('" + "','".join(res) + "')"};
			"""
			logger_.info("同步user_info表数据————————————————————————————————")
			pdata = pd.read_sql(text(sql), nebula_engine.connect())
			pdata.to_sql("user_info", test_nebula_engine, if_exists="append", index=False)
			# 其他用户表同步
			for tn in nebula_table:
				sql = f"""
					select * from {tn} where
					user_id in {f"('" + "','".join(res) + "')"};
				"""
				logger_.info(f"同步{tn}表数据————————————————————————————————")
				pdata = pd.read_sql(text(sql), nebula_engine.connect())
				pdata.to_sql(tn, test_nebula_engine, if_exists="append", index=False)
